/*
 * 文件名：ProtocalConnectMgrServiceImpl.java
 * 版权：亚信联创版权所有
 * 描述：TODO
 * 修改人：xuwei3
 * 修改时间：2015-5-20
 * 修改内容：TODO
 */
package com.ailk.bigdata.etl.realstream.server.protocal.sdtp.server.impl;

import java.nio.charset.Charset;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;

import com.ailk.bigdata.etl.common.server.net.NettyUtil;
import com.ailk.bigdata.etl.realstream.server.constant.ProtocalSdtpConstant;
import com.ailk.bigdata.etl.realstream.server.dao.RealsocksUserDao;
import com.ailk.bigdata.etl.realstream.server.dao.impl.RealsocksUserDaoImpl;
import com.ailk.bigdata.etl.realstream.server.model.MessageType;
import com.ailk.bigdata.etl.realstream.server.model.RealStreamUser;
import com.ailk.bigdata.etl.realstream.server.model.RealsocksUser;
import com.ailk.bigdata.etl.realstream.server.model.SdtpDataPack;
import com.ailk.bigdata.etl.realstream.server.monitor.EventCounter;
import com.ailk.bigdata.etl.realstream.server.protocal.sdtp.server.ProtocalConnectMgrService;
import com.ailk.bigdata.etl.realstream.server.tools.RealStreamTool;

/**
 * @description TODO
 * @author [xuwei3]
 * @version [版本号,2015-5-20]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public class ProtocalConnectMgrServiceImpl implements ProtocalConnectMgrService
{
	private static Logger logger = LoggerFactory.getLogger(ProtocalConnectMgrServiceImpl.class);
	private static Charset utf8Charset = Charset.forName("utf-8");

	private EventCounter eventCounter;

	public ProtocalConnectMgrServiceImpl(EventCounter eventCounter)
	{
		this.eventCounter = eventCounter;
	}

	/**
	 * @see com.ailk.bigdata.etl.realstream.sdtp.server.ProtocalConnectMgrService#versionNego()
	 */
	@Override
	public void versionNego(ChannelHandlerContext ctx, SdtpDataPack dataPack)
	{
		dataPack.getHeader().setMessageType(MessageType.verNego_Resp.getValue());
		dataPack.getHeader().setTotalLength(10);
		ByteBuf dataContent = dataPack.getDataContent();
		short version = dataContent.readUnsignedByte();
		short subVersion = dataContent.readUnsignedByte();
		dataContent.clear();

		short versionPassed = ProtocalSdtpConstant.SDTP_PROTOCAL_VERSION_PASSED;

		logger.info("主版本[{}]子版本[{}]", version, subVersion);
		if (ProtocalSdtpConstant.SDTP_PROTOCAL_VERSION_PAR != version || ProtocalSdtpConstant.SDTP_PROTOCAL_VERSION_SUB != subVersion)
		{
			if (ProtocalSdtpConstant.SDTP_PROTOCAL_VERSION_PAR < version)
			{
				versionPassed = ProtocalSdtpConstant.SDTP_PROTOCAL_VERSION_OVER;
			}
			else if (ProtocalSdtpConstant.SDTP_PROTOCAL_VERSION_PAR > version)
			{
				versionPassed = ProtocalSdtpConstant.SDTP_PROTOCAL_VERSION_UNDER;
			}
			else if (ProtocalSdtpConstant.SDTP_PROTOCAL_VERSION_SUB > subVersion)
			{
				versionPassed = ProtocalSdtpConstant.SDTP_PROTOCAL_VERSION_OVER;
			}
			else
			{
				versionPassed = ProtocalSdtpConstant.SDTP_PROTOCAL_VERSION_UNDER;
			}
		}

		if (versionPassed == ProtocalSdtpConstant.SDTP_PROTOCAL_VERSION_PASSED)
		{

			String clientAdd = NettyUtil.getClientAdd(ctx);
			RealStreamUser user = new RealStreamUser();
			user.setClientIP(clientAdd);// 客户端地址信息
			user.setCtx(ctx);
			RealStreamTool.AUTHENTICATION_INFO_CACHE.put(clientAdd, user);

		}

		dataContent.writeByte(versionPassed);

		logger.info("版本协商请求协商结果["
				+ (versionPassed == ProtocalSdtpConstant.SDTP_PROTOCAL_VERSION_PASSED ? "通过"
						: (versionPassed == ProtocalSdtpConstant.SDTP_PROTOCAL_VERSION_OVER ? "版本过高" : "版本过低")) + "].........");

		ctx.writeAndFlush(dataPack);
	}

	/**
	 * @see com.ailk.bigdata.etl.realstream.sdtp.server.ProtocalConnectMgrService#linkAuth()
	 */
	@Override
	public void linkAuth(ChannelHandlerContext ctx, SdtpDataPack dataPack)
	{
		String clientAddr = NettyUtil.getClientAdd(ctx);
		//String clientIp = clientAddr.split(":")[0];
		if (!RealStreamTool.AUTHENTICATION_INFO_CACHE.containsKey(clientAddr))
		{
			logger.error("版本未协商,客户端连接失败[{}]..............", clientAddr);
			ctx.close();
			return;
		}

		dataPack.getHeader().setMessageType(MessageType.linkAuth_Resp.getValue());
		dataPack.getHeader().setTotalLength(74);

		RealsocksUserDao userDao = new RealsocksUserDaoImpl();
		ByteBuf dataContent = dataPack.getDataContent();

		String userId = new String(dataContent.readBytes(12).array(), utf8Charset).trim();
		//String digest = new String(dataContent.readBytes(64).array(), utf8Charset).trim();
		byte[] digestbyte = dataContent.readBytes(64).array();
		StringBuilder sb2 = new StringBuilder();
		for(int i = 0;i < digestbyte.length;i++){
			sb2.append(String.format("%02X", digestbyte[i]));
		}
		String digest = sb2.toString().substring(0, 64);
		
		long timestamp = dataContent.readUnsignedInt();
		int rand = dataContent.readUnsignedShort();
		RealsocksUser realSockUser = userDao.findByUserName(userId);

		if (null != realSockUser)
		{
			realSockUser.setTimestamp(timestamp);
			realSockUser.setRand(rand);
		}

		dataContent.clear();

		if (null == realSockUser)
		{
			dataContent.writeByte(ProtocalSdtpConstant.SDTP_PROTOCAL_LOGIN_AUTH_IDNOTEXIST);
			logger.error("用户不存在，认证失败........................");
		}
		else if (StringUtils.isNotEmpty(digest) && digest.equalsIgnoreCase(realSockUser.getDigest()))
		{
			dataContent.writeByte(ProtocalSdtpConstant.SDTP_PROTOCAL_LOGIN_AUTH_PASS);
			dataContent.writeBytes(digestbyte);

			RealStreamUser user = new RealStreamUser();
			user.setClientIP(clientAddr);// 客户端地址信息
			user.setCtx(ctx);
			if (!RealStreamTool.AUTHORISED_USER_INFO.containsKey(clientAddr)) {
				RealStreamTool.AUTHORISED_USER_INFO.put(clientAddr, user);
			}
			RealStreamTool.AUTHENTICATION_INFO_CACHE.remove(clientAddr);
			logger.info("用户{}认证通过........................", realSockUser.getUserAcct());
		}
    		else
    		{
    
    			dataContent.writeByte(ProtocalSdtpConstant.SDTP_PROTOCAL_LOGIN_AUTH_SHA_ENCRYPT_ERR);
    			dataContent.writeBytes(realSockUser.getDigest().getBytes(utf8Charset));
    			logger.error("用户[{}]认证码校验错误，服务器getdigest[{}],客户端digest[{}]认证失败........................", realSockUser.getUserAcct(),
    				realSockUser.getDigest(),digest);
    
    		}
    
    		ctx.writeAndFlush(dataPack);
	}

	/**
	 * @see com.ailk.bigdata.etl.realstream.sdtp.server.ProtocalConnectMgrService#linkCheck()
	 */
	@Override
	public void linkCheck(ChannelHandlerContext ctx, SdtpDataPack dataPack)
	{
		// server端接收到链路检测请求后立即返回，无内容
		dataPack.getHeader().setMessageType(MessageType.linkCheck_Req.getValue());

		ctx.writeAndFlush(dataPack);
	}

	/**
	 * @see com.ailk.bigdata.etl.realstream.sdtp.server.ProtocalConnectMgrService#linkDataCheck()
	 */
	@Override
	public void linkDataCheck(ChannelHandlerContext ctx, SdtpDataPack dataPack)
	{
		String clientAddr = NettyUtil.getClientAdd(ctx);
		//String clientIp = clientAddr.split(":")[0];
		ByteBuf dataContent = dataPack.getDataContent();
		dataPack.getHeader().setTotalLength(9 + 13);
		int sendFlag = dataContent.readInt();
		int sendDataInfo = dataContent.readInt();
		dataContent.clear();

		int acceptedCount = RealStreamTool.AUTHORISED_USER_INFO.get(clientAddr).getAcceptedAndReset();

		dataContent.writeInt(sendFlag);
		int dataEqual = ProtocalSdtpConstant.SDTP_PROTOCAL_DATACHECK_LACK;
		if (acceptedCount == sendDataInfo)
		{
			dataEqual = ProtocalSdtpConstant.SDTP_PROTOCAL_DATACHECK_EQUAL;
		}
		else if (acceptedCount > sendDataInfo)
		{
			dataEqual = ProtocalSdtpConstant.SDTP_PROTOCAL_DATACHECK_BIGGER;
		}

		if (logger.isInfoEnabled())
		{
			logger.info("客户端[{}]发起数据条数[{}],服务端接收条数[{}]", clientAddr, sendDataInfo, acceptedCount);
			logger.info("客户端[{}]发起数据校验请求,校验结果[{}].........", clientAddr, (dataEqual == ProtocalSdtpConstant.SDTP_PROTOCAL_DATACHECK_EQUAL ? "数据量正确"
					: (dataEqual == ProtocalSdtpConstant.SDTP_PROTOCAL_DATACHECK_LACK ? "接收方数据小于发送方数据" : "接收方数据大于发送方数据")));
		}

		dataContent.writeByte(dataEqual);
		dataContent.writeInt(sendDataInfo);
		dataContent.writeInt(acceptedCount);

		ctx.writeAndFlush(dataPack);

	}

	/**
	 * @see com.ailk.bigdata.etl.realstream.sdtp.server.ProtocalConnectMgrService#linkRel()
	 */
	@Override
	public void linkRel(ChannelHandlerContext ctx, SdtpDataPack dataPack)
	{
		String clientAddr = NettyUtil.getClientAdd(ctx);
		//String clientIp = clientAddr.split(":")[0];
		logger.info("客户端断开连接[{}]............", clientAddr);
		RealStreamTool.AUTHORISED_USER_INFO.remove(clientAddr);
		RealStreamTool.ACCEPTING_CHANNEL_CACHE.remove(clientAddr);

		ctx.writeAndFlush(dataPack).addListener(ChannelFutureListener.CLOSE);
	}

}
